package com.itcam.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

import java.util.Properties;

/**
 * @author : Cammy.Wu
 * Description : 
 */

@SpringBootTest(classes = ProducerApplication.class)
public class JavaProducerDemo {

    // 主题名称
    public static final String TOPIC_NAME = "hello_kafka";

    /**
     * 测试方法，用于向 Kafka 发送消息
     */
    @Test
    public void sendMsg() {
        // 创建 Properties 对象，用于存储 Kafka 生产者的配置信息
        Properties properties = new Properties();
        // 设置 Kafka 集群的初始连接地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.150.101:9092");
        // 设置消息键的序列化类，将键序列化为字符串
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 设置消息值的序列化类，将值序列化为字符串
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // 根据配置信息创建 Kafka 生产者对象
        KafkaProducer kafkaProducer = new KafkaProducer(properties);

        // 循环 10 次，向 Kafka 发送 10 条消息
        for (int i = 0; i < 10; i++) {
            // 创建一个生产者记录对象，指定消息要发送到的主题和消息内容
            ProducerRecord producerRecord = new ProducerRecord(TOPIC_NAME, "hello_kafka!" + i);
            // 调用生产者的 send 方法，将消息发送到 Kafka
            kafkaProducer.send(producerRecord);
        }

        // 关闭 Kafka 生产者，释放资源
        kafkaProducer.close();
    }

}
